package com.leilei;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.util.Arrays;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @author lei
 * @version 1.0
 * @date 2021/3/21 13:21
 */
public class Flink_Table_2_demo {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        DataStreamSource<GirlFriend> scFriendList = env.fromCollection(Arrays.asList(new GirlFriend("小丽", "A", 22),
                new GirlFriend("小丽于", "C", 24),
                new GirlFriend("小胡1", "D", 23),
                new GirlFriend("小丽于", "D", 17),
                new GirlFriend("小胡2", "D", 25)));
        Table scTable = tableEnv.fromDataStream(scFriendList);
        Table table = scTable.select($("name"), $("cup"), $("age"))
                .filter($("age").isGreater(18).and($("name").like("%小%")))
                .orderBy($("age").desc())
                .limit(1, 3);
        DataStream<Tuple2<Boolean, GirlFriend>> toRetractStream = tableEnv.toRetractStream(table, GirlFriend.class);
        toRetractStream.filter(t->t.f0).print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


}
